import base64
from urllib import parse
import zlib
import numpy as np
from pymongo import MongoClient
import pymysql
import math
from concurrent.futures import ThreadPoolExecutor, as_completed

from utilsapp.common import  get_collection,get_collection_name
import json
import random
import time
from datetime import datetime, timedelta
import requests


def algorithm(config, sleep_secs,app):
    if 'algorithm' not in config['business'] or not config['business']['algorithm']:
        print('算法引擎未开启')
        return
    sleep_secs = 60 * config['business']['algorithm_time'] if 'algorithm_time' in config['business'] else sleep_secs
    while True:
        try:
            print('算法引擎开始时间:' + datetime.now().strftime('%H:%M:%S'))
            run(config,app)
            print('算法引擎结束时间:' + datetime.now().strftime('%H:%M:%S'))
        except Exception as e:
            print('算法引擎异常', e)
        finally:
            time.sleep(sleep_secs)

def run(settings,app):
    con = pymysql.connect(host=settings['mysql']["host"], user=settings['mysql']["user"], password=settings['mysql']["password"],
                          database=settings['mysql']["dbname"], port=(settings['mysql']["port"]))
    mgc = settings["mongodb"]
    user = mgc.get("user", "").strip()
    password = mgc.get("password", "").strip()
    if password:
        connection_string = (
            f"mongodb://{parse.quote_plus(user)}:"
            f"{parse.quote_plus(password)}@"
            f'{mgc["host"]}:{mgc["port"]}/'
            )
    else:
        connection_string = f'mongodb://{mgc["host"]}:{mgc["port"]}/'
    if 'url' in mgc:
        connection_string = mgc['url']
    mgdb = MongoClient(connection_string)
    headers = {"Content-Type": "application/json"}
    login_data = {"username": "admin_end", "password": "78a8612541876276f1c8d5de52328f69"}
    url = settings['business']['algorithm_url'] if 'algorithm_url' in settings['business'] else 'http://59.172.2.139:8088'
    access_token = post(url,"/login", login_data, headers)['access_token']
    headers['Authorization'] = access_token
    models = get(url,"/restful/cal/settings/models", login_data, headers)
    db_insert_gx('algorithm_model', models, con)
    execute(mgdb,con,headers,access_token,url,settings,app)
    return

def process_item(item, mgdb, headers, access_token, url, settings, app):
    """处理单个算法配置项"""
    if item['is_enable'] == 0:
        return
    
    # 为每个线程创建独立的数据库连接
    con = pymysql.connect(
        host=settings['mysql']["host"], 
        user=settings['mysql']["user"], 
        password=settings['mysql']["password"],
        database=settings['mysql']["dbname"], 
        port=settings['mysql']["port"]
    )
    
    try:
        redis = app.redis
        sensor_id = item['sensor_id']
        sensor = db_connect(f'''select s.*,s.status,g.proj_no,g.SignName,g.name as group_name,f.name as factory_name,z.name as zone_name,
                            d.device_name,d.group_id,d.company_id,d.factory_id,d.zone_id from sensor s 
                            LEFT JOIN regionlevel z on z.id=s.zone_id 
                            LEFT JOIN regionlevel f on f.id=s.factory_id 
                            LEFT JOIN regionlevel g on g.id = s.group_id 
                            LEFT JOIN device d on d.id = s.device_id 
                            where s.id = {sensor_id}''',con)[0]
        print(sensor['status'])
        if sensor['status'] != '在线':
            return
        
        proj_no = sensor['proj_no']
        sport = sensor['port']
        saddr = sensor['address'].split('_')[0]
        
        channel = sensor['address'].split('_')[1] if len(sensor['address'].split('_')) == 2 else 0
        
        etime = int(datetime.now().timestamp())
        models = json.loads(item['config'])
        models = sorted(models, key=lambda model: model['m_type'] not in ['动态阈值','固定阈值','趋势预测'])
        isWarn = False
        for model in models:
            if model['m_type'] in ['动态阈值','固定阈值','趋势预测']:
                mongb1 = mgdb[f"ieplat_{proj_no}"]
                dbname = f"eige_{sport}_{saddr.split('_')[0]}"
                dbname1 = get_collection_name(mongb1,dbname,redis)
                collection = mongb1[dbname1]
                for eigkey, config in model['m_config'].items():
                    axial = next((item.get('value') for item in config if item['key'] == 'axial'), 'Z') or 'Z'
                    stime = int((datetime.now() - timedelta(minutes=int(next(item for item in config if item['key'] == 'data_time_series')['value']))).timestamp())
                    query = {
                        "$and": [
                            {"_id": {"$gte": f"{sport}_{saddr}_{channel}_{stime}"}},
                            {"_id": {"$lte": f"{sport}_{saddr}_{channel}_{etime}"}},
                        ]
                    }
                    if eigkey in ['tem', 'a_cur', 'b_cur', 'c_cur']:
                        projection = { "_id":0, f"data.{eigkey}": 1}
                        result = collection.find(query, projection)
                        signal = [doc.get('data', {}).get(eigkey) for doc in result if doc.get('data', {}).get(eigkey) is not None]
                    else:
                        projection = { "_id":0, f"data.{axial}.{eigkey}": 1}
                        result = collection.find(query, projection)
                        signal = [doc.get('data', {}).get(axial, {}).get(eigkey) for doc in result if doc.get('data', {}).get(axial, {}).get(eigkey) is not None]

                    if len(signal) == 0:
                        continue
                    code = f"{model['m_m_code']}_{model['m_version']}".replace('.', '_')
                    data = {
                        "data_info":{},
                        "token": access_token,
                        "data": {
                            "signal1": signal,
                        },
                        "data_result":{
                            "s1_result": []
                        },
                        "config":{
                            "m_m_code": model['m_m_code'],
                            "version": model['m_version'], 
                            "m_config":model['m_config'][eigkey]
                        }
                    }
                    type = 'tr'
                    res = post(url,f"/restful/cal/{type}/{code}", data, headers)
                    if 'data_result' not in res:
                        print(res)
                        continue
                    model['m_config'][eigkey] = res['config']['m_config']
                    isWarn = sent_warn(item, sensor, model, eigkey, res, datetime.now().strftime('%Y-%m-%d %H:%M:%S'), con, settings) or isWarn
            if model['m_type'] in ['机理模型','AI模型'] and isWarn:
                collection = mgdb[f"ieplat_{proj_no}"][f"wave"]
                for eigkey, config in model['m_config'].items():
                    axial = next((item.get('value') for item in config if item['key'] == 'axial'), 'Z') or 'Z'
                    stime = int((datetime.now() - timedelta(hours=24)).timestamp())
                    query = {"_id": {"$gte": f"{sport}_{saddr}_{channel}_{axial}_{stime}", "$lte": f"{sport}_{saddr}_{channel}_{axial}_{etime}"}}
                    pipeline_latest = [
                        {"$match": query},
                        {"$sort": {"_id": -1}},
                        {"$project": {
                            "_id": 1,
                            "data": 1
                        }},
                    ]
                    result = list(collection.aggregate(pipeline_latest))
                    if(len(result) == 0):
                        continue
                    signal = get_wave(result[0]['data']['acc'][eigkey])
                    now = datetime.fromtimestamp(int(result[0]['_id'].split('_')[-1])).strftime('%Y-%m-%d %H:%M:%S')
                    fs =  result[0]['data']['freq']
                    data = {
                        "data_info":{},
                        "token": access_token,
                        "data": {
                            "signal1": signal,
                            "s1_fs": fs
                        },
                        "data_result":{
                            "s1_result": [],
                        },
                        "config":{
                            "m_m_code": 'ON_OFF_FILTER',
                            "version": 'V1.0', 
                            "m_config": []
                        }
                    }
                    res = post(url,f"/restful/cal/fu/ON_OFF_FILTER_V1_0", data, headers)
                    if 'data_result' not in res:
                        print('ON_OFF_FILTER算法接口出错')
                        continue
                    if res['data_result']['s1_result'][0] == 0:
                        print(sensor['device_name'] + '停机')
                        continue
                    code = f"{model['m_m_code']}_{model['m_version']}".replace('.', '_')
                    data = {
                        "data_info":{},
                        "token": access_token,
                        "data": {
                            "signal1": signal,
                            "s1_fs": fs
                        },
                        "data_result":{
                            "s1_result": [],
                        },
                        "config":{
                            "m_m_code": model['m_m_code'],
                            "version": model['m_version'], 
                            "m_config":model['m_config'][eigkey]
                        }
                    }
                    type = 'ai' if model['m_type'] == 'AI模型' else 'ma'
                    res = post(url,f"/restful/cal/{type}/{code}", data, headers)
                    sent_warn(item,sensor,model,eigkey,res,now,con,settings)
        
        item['config'] = json.dumps(models)
        db_update('algorithm_config', item, f"WHERE id=%s", {"id": item['id']}, con)
    finally:
        con.close()

def execute(mgdb,con,headers,access_token,url,settings,app):
    """并行执行算法配置项处理"""
    items = db_connect('select * from algorithm_config',con)
    
    # 使用线程池并行处理items
    max_workers = min(10, len(items))  # 限制最大线程数，避免资源过度消耗
    with ThreadPoolExecutor(max_workers=max_workers) as executor:
        # 提交所有任务
        futures = {
            executor.submit(process_item, item, mgdb, headers, access_token, url, settings, app): item 
            for item in items
        }
        
        # 等待所有任务完成
        for future in as_completed(futures):
            try:
                future.result()  # 获取结果，如果有异常会在这里抛出
            except Exception as e:
                item = futures[future]
                print(f"处理算法配置项 {item.get('id', 'unknown')} 时发生错误: {e}")

def get_wave(wave):
    if type(wave) is list:
        return wave

    # 将解压缩后的字节数据转换回浮点数组
    restored_data = np.frombuffer(zlib.decompress(wave), dtype=np.float32).tolist()
    return [round(num, 4) for num in restored_data]
        
def sent_warn(item,sensor,model,eigkey,res,now,con,settings):
    if 'data_result' not in res or len(res['data_result']['s1_result']) < 3:
        print(model['m_m_code'] + '算法接口出错')
        return False
    dir = {
        'acc_rms': '加速度',
        'vel_rms': '速度',
        'tem': '温度',
        'acc_kur': '加速度峭度',
        'acc_wave': '加速度波形',
    }
    unit = {
        'acc_rms': 'm/s²',
        'vel_rms': 'mm/s',
        'tem': '°C',
        'acc_kur': '',
        'acc_wave': '',
    }
    print(sensor['device_name'],sensor['sensor_name'],model['m_type'],eigkey,res['data_result']['s1_result'])
    warn_level = res['data_result']['s1_result'][0]
    if warn_level > 0:
        warns = db_connect(f"select * from algorithm_warn where is_show = 1 and m_m_code = '{model['m_m_code']}' and sensor_id = '{item['sensor_id']}' and eigkey = '{eigkey}'",con)
        if(len(warns) == 0):
            warn = {
                'content': f"{dir[eigkey]}|{model['m_desc']}",
                'm_m_code': model['m_m_code'],
                'eigkey': eigkey,
                'device_id': item['device_id'],
                'sensor_id': item['sensor_id'],
                'notice': 0,
                'warn_level': warn_level,
                'standard_data': res['data_result']['s1_result'][2],
                'pred_data': res['data_result']['s1_result'][1],
                'pred_time': now,
                'm_config': json.dumps(res['config']['m_config']),
                'create_time': now,
                'is_show': 1
            }
            db_insert('algorithm_warn', warn, con)
        else:
            warn = warns[0]
            db_update('algorithm_warn', {**warn, **{
                'notice': 0,
                'warn_level': warn_level,
                'standard_data': res['data_result']['s1_result'][2],
                'pred_data': res['data_result']['s1_result'][1],
                'pred_time': now,
                'm_config': json.dumps(res['config']['m_config']),
            }}, f"WHERE id=%s", {"id": warn['id']}, con)
        content = f"{sensor['device_name']}|{sensor['sensor_name']}|{dir[eigkey]}|{model['m_desc']}"
        notices = db_connect(f"select * from algorithm_notice where content like '{content}'",con)
        if len(notices) == 0:
            notice = {
                'content': content,
                'device_id': item['device_id'],
                'sensor_id': item['sensor_id'],
                'sent_time': now,
                'handle': 0,
                'type': 'warn'
            }
            db_insert('algorithm_notice', notice, con)
        else:
            notice = notices[0]
            db_update('algorithm_notice', {**notice, **{
                'handle': 0,
                'sent_time': now,
            }}, f"WHERE id=%s", {"id": notice['id']}, con)
        if 'algorithm_sync' in settings['business'] and settings['business']['algorithm_sync']:
            sync_content = f"{sensor['device_name']}|{sensor['sensor_name']}|{dir[eigkey]}{model['m_desc']}|Z_{eigkey}"
            sync_warns = db_connect(f"select * from warn where is_show = 1 and content = '{sync_content}' and sensor_id = '{item['sensor_id']}'",con)
            wData = {
                'cause': None,
                'suggest': None,
            }
            if('algorithm_url' not in settings['business']):
                wData = json.loads(requests.post("http://phm.hbicc.com/analysisWarn", data=json.dumps({
                    'device_name': sensor['device_name'],
                    'sensor_name': sensor['sensor_name'],
                    'data_type': dir[eigkey],
                    'level_str': "轻微" if warn_level == 1 else "中等" if warn_level == 2 else "严重",
                    'description': model['m_desc']
                }), headers={"Content-Type": "application/json"}).text)
            if(len(sync_warns) == 0):
                sync_warn = {
                    'title': model['m_type'],
                    'content': sync_content,
                    'warn_source': model['m_type'],
                    'warn_type': '1',
                    'warn_level': warn_level,
                    'group_id': sensor['group_id'],
                    'company_id': sensor['company_id'],
                    'factory_id': sensor['factory_id'],
                    'zone_id': sensor['zone_id'],
                    'device_id': item['device_id'],
                    'sensor_id': item['sensor_id'],
                    'type_id': -1,
                    'notice': 0,
                    'create_time': now,
                    'is_show': 1,
                    'is_sms': 0,
                    'is_handled': 0,
                    'standard_data': res['data_result']['s1_result'][2],
                    'pred_data': res['data_result']['s1_result'][1],
                    'pred_time': now,
                    'cause': wData['cause'],
                    'suggest': wData['suggest'],
                    'config': '{}',
                    'istest': 2,
                }
                db_insert('warn', sync_warn, con)
            else:
                sync_warn = sync_warns[0]
                db_update('warn', {**sync_warn, **{
                    'notice': 0,
                    'warn_level': warn_level,
                    'standard_data': res['data_result']['s1_result'][2],
                    'pred_data': res['data_result']['s1_result'][1],
                    'pred_time': now,
                    'cause': wData['cause'],
                    'suggest': wData['suggest'],
                }}, f"WHERE id=%s", {"id": sync_warn['id']}, con)

            json_data = json.dumps({
                "type_": 1,
                "device_ids": [item['device_id']],
                "times": [(datetime.strptime(now, "%Y-%m-%d %H:%M:%S") - timedelta(days=1)).strftime('%Y-%m-%d %H:%M:%S'),now]
            })  # 设置请求头  # 发送POST请求
            headers = {'Content-Type': 'application/json'}
            requests.post(f'http://127.0.0.1:5600/api/report/down/sys', data=json_data, headers=headers)
        if('algorithm_url' not in settings['business']):
            userList = db_connect(f"""SELECT userinfo.*, role.dataId FROM userinfo JOIN role ON userinfo.role_id = role.id WHERE phonenumber IS NOT NULL""",con)
            mailList = []
            phoneList = []
            for item in userList:
                if warn_level >= item['message'] and (
                        item['dataId'] == 0 or sensor['group_id'] == item['dataId'] or sensor['company_id'] == item['dataId'] or
                        sensor['factory_id'] == item['dataId'] or sensor['zone_id'] == item['dataId']):
                    mailList.append('+86' + item['phonenumber'])
                if warn_level >= item['mail'] and (
                        item['dataId'] == 0 or sensor['group_id'] == item['dataId'] or sensor['company_id'] == item['dataId'] or
                        sensor['factory_id'] == item['dataId'] or sensor['zone_id'] == item['dataId']):
                    phoneList.append(item['phonenumber'])
            warnData = {
                'zone': sensor['factory_name'] + sensor['zone_name'],
                'device_name': sensor['device_name'],
                'sensor_name': sensor['sensor_name'],
                'SignName': sensor.get('SignName') or "武汉中云康崇",
                'standard_data': res['data_result']['s1_result'][2],
                'pred_data': res['data_result']['s1_result'][1],
                'warn_level': warn_level,
                'content': f"{dir[eigkey]}|{model['m_desc']}",
                'pred_time': now,
                'unit': unit[eigkey]
            }
            requests.post("http://phm.hbicc.com/warn1", data=json.dumps({
                'mailList': mailList,
                'phoneList': phoneList,
                'data':warnData
            }), headers={"Content-Type": "application/json"})
        return True
    return False


def db_connect(sql, _conn):
    cur = _conn.cursor()
    try:
        for i in range(2):
            try:
                cur.execute(sql)
                datas = cur.fetchall()
                fields = [field[0] for field in cur.description]
                res = [dict(zip(fields, result)) for result in datas]
                return res
            except Exception as e:
                return e

    finally:
        cur.close()
def db_update(tableName, model, where_clause, where_model, _conn):
    cur = _conn.cursor()
    try:
        # 排除需要修改的字段,默认添加id字段,前端可以在model传入not_db_keys
        not_db_keys = model['not_db_keys'] + ['id', 'not_db_keys'] if type(
            model) == dict and 'not_db_keys' in model else ['id']
        db_model = {key: model[key] for key in model if key not in not_db_keys}
        # 列的字段
        set_clause = ', '.join(
            [f"`{column}` = %s" for column in db_model.keys()])

        sql = f"UPDATE {tableName} SET {set_clause} {where_clause}"
        cur.execute(sql, tuple(list(db_model.values()) +
                               list(where_model.values())))
        _conn.commit()
        return {"Affected_rows": cur.rowcount, 'msg': '修改成功', 'code': 0, 'success': True, }
    except Exception as e:
        return {"Affected_rows": 0, 'error': str(e), 'msg': '修改失败', 'code': -1, 'success': False, }
    finally:
        cur.close()

def db_insert(tableName, adddataList, _conn):
    cur = _conn.cursor()
    try:
        dataList = [adddataList] if type(adddataList) == dict else adddataList
        add_models = []
        for item in dataList:
            model = {key: item[key] for key in item }
            add_models.append({k: model[k] for k in sorted(model)})
        columns = '`, `'.join(add_models[0].keys())
        values = ', '.join(['%s'] * len(add_models[0]))
        sql = f"INSERT INTO `{tableName}` (`{columns}`) VALUES ({values}) "
        params = [tuple(row.values()) for row in add_models]
        cur.executemany(sql, params)
        _conn.commit()
        return {"Affected_rows": cur.rowcount,  'msg': '新增成功', 'code': 0, 'success': True, }
    except Exception as e:
        return {"Affected_rows": 0, 'error': str(e), 'msg': '新增失败', 'code': -1, 'success': False, }
    finally:
        cur.close()

def db_insert_gx(tableName, adddataList, _conn):
    cur = _conn.cursor()
    try:
        dataList = [adddataList] if type(adddataList) == dict else adddataList
        add_models = []
        for item in dataList:
            model = {key: item[key] for key in item }
            add_models.append({k: model[k] for k in sorted(model)})
        columns = '`, `'.join(add_models[0].keys())
        values = ', '.join(['%s'] * len(add_models[0]))
        sql = f"INSERT INTO `{tableName}` (`{columns}`) VALUES ({values}) "
        params = [tuple(row.values()) for row in add_models]
        cur.execute(f"TRUNCATE TABLE `{tableName}`")
        cur.executemany(sql, params)
        _conn.commit()
        return {"Affected_rows": cur.rowcount,  'msg': '新增成功', 'code': 0, 'success': True, }
    except Exception as e:
        return {"Affected_rows": 0, 'error': str(e), 'msg': '新增失败', 'code': -1, 'success': False, }
    finally:
        cur.close()

def get(base,url,data,headers):
    return json.loads(requests.get(base + url, data=json.dumps(data), headers=headers).text)

def post(base,url,data,headers):
    return json.loads(requests.post(base + url, data=json.dumps(data), headers=headers).text)